#pragma once
#include "../comm/Log.h"
#include "../comm/util.h"
#include "../comm/msg.pb.h"
#include <unordered_map>
#include <memory>
#include <list>
#include <mutex>

namespace ns_message
{
    using namespace ns_log;
    using namespace ns_util;
    using namespace ns_proto;

    // 消息的数据属性描述
    using MessagePtr = shared_ptr<ns_proto::Message>;

    // 消息的文件式持久化处理
    class MessageDurable
    {
    public:
        MessageDurable(string path, const string &qname)
            : _qname(qname)
        {
            // 初始化文件名
            if (path.back() != '/') // 防止
                path += '/';
            _msgfile = path + qname + ".msg";
            _msgtmpfile = path + qname + ".tmpmsg";
        }
        ~MessageDurable()
        {
            // 删除临时文件
            if (FileOper::exit(_msgtmpfile))
                assert(FileOper::removefile(_msgtmpfile));
        }

        // 创建和删除消息文件
        void create_msgfile()
        {
            // 不用创建临时文件，因为临时文件只用于垃圾回收
            assert(FileOper::createfile(_msgfile));
        }
        void delete_msgfile()
        {
            assert(FileOper::removefile(_msgfile));
        }

        // 添加和删除消息
        void insert_msg(MessagePtr &msgp)
        {
            // 向文件中写入数据信息
            _insert(_msgfile, msgp);
        }
        void delete_msg(const MessagePtr &msgp)
        {
            // 因为文件中存放的是二进制的序列化的消息
            // 因此采用先进行同一份数据的valid，然后进行覆盖原数据
            // 1.修改数据的有效标志
            msgp->mutable_payload()->set_valid("0");

            // 2.序列化数据格式
            string str = msgp->payload().SerializeAsString();
            if (str.size() != msgp->len())
            {
                LOG(WARNING) << "需要删除的数据不存在" << endl;
                return;
            }

            // 3.将数据添加进相同偏移量的位置
            bool ret = FileOper::write(_msgfile, &str[0], msgp->offset(), str.size());
            if (!ret)
            {
                LOG(WARNING) << "文件写入失败" << endl;
                return;
            }
        }

        // 数据恢复+垃圾回收(Garbage_collection)
        list<MessagePtr> recovery_and_gc()
        {
            // 1.读取文件中的有效消息到链表中
            list<MessagePtr> result;
            _load_message(_msgfile, result);

            // 2.将链表中的数据写进临时文件中
            assert(FileOper::createfile(_msgtmpfile));
            for (auto &msgp : result)
            {
                // 向文件中写入数据信息
                _insert(_msgtmpfile, msgp);
            }

            bool ret;
            // 3.删除原文件并将临时文件重命名
            ret = FileOper::removefile(_msgfile);
            if (!ret)
            {
                LOG(WARNING) << "删除文件失败" << endl;
                return result;
            }
            ret = FileOper::rename(_msgtmpfile, _msgfile);
            if (!ret)
            {
                LOG(WARNING) << "修改文件名称失败" << endl;
                return result;
            }

            return result;
        }

    private:
        // 向指定文件中写入数据信息
        void _insert(const string &filename, MessagePtr &msgp)
        {
            // 1.序列化数据格式
            string str = msgp->payload().SerializeAsString();

            // 2.采用(len+data)方式将数据添加进文件末尾
            size_t fsize = FileOper::size(filename); // 指向文件结尾
            size_t len = str.size();
            bool ret = FileOper::write(filename, (char *)&len, fsize, sizeof(size_t));
            if (!ret)
            {
                LOG(WARNING) << "信息长度写入失败" << endl;
                return;
            }
            fsize += sizeof(size_t); // 偏移到最后
            ret = FileOper::write(filename, &str[0], fsize, str.size());
            if (!ret)
            {
                LOG(WARNING) << "信息数据写入失败" << endl;
                return;
            }

            // 3.更新Message中的偏移量和文件长度
            msgp->set_offset(fsize);
            msgp->set_len(str.size());
        }

        // 加载文件中的的有效数据
        void _load_message(string &file_name, list<MessagePtr> &result)
        {
            size_t offset = 0; // 文件的偏移位置
            int file_size = FileOper::size(file_name);
            bool ret;
            while (offset < file_size)
            {
                size_t len;
                // 1.1获取长度信息
                ret = FileOper::read(file_name, (char *)&len, offset, sizeof(size_t));
                if (!ret)
                {
                    LOG(WARNING) << "文件读取失败" << endl;
                    return;
                }
                offset += sizeof(size_t);

                // 1.2读取信息内容串
                string str;
                str.resize(len);
                ret = FileOper::read(file_name, &str[0], offset, len);
                if (!ret)
                {
                    LOG(WARNING) << "文件读取失败" << endl;
                    return;
                }
                offset += len;

                // 1.3.反序列化文件消息并构建对象添加进链表(如果有效)
                MessagePtr msgp = make_shared<ns_proto::Message>();
                // msgp->set_len(len);
                // msgp->set_offset(offset);
                msgp->mutable_payload()->ParseFromString(str); // 反序列化
                if (msgp->payload().valid() == "1")
                    result.push_back(msgp);
            }
        }

    private:
        string _qname;      // 队列名称(每一个消息文件所属的队列)
        string _msgfile;    // 消息文件
        string _msgtmpfile; // 消息临时文件(用于垃圾消息回收)
    };

    // 以队列为单元对消息进行管理
    class QueueMessage
    {
    public:
        QueueMessage(string path, const string &qname)
            : _qname(qname), _message_durable(path, qname),
              _durable_total_size(0), _durable_valid_size(0)
        {
            // 创建当前队列的二进制文本文件
            _message_durable.create_msgfile();
        }

        void _recovery()
        {
            unique_lock<mutex> lock(_mtx);
            // 文件的消息数据恢复
            _push_messages = _message_durable.recovery_and_gc();
            for (const auto &msgp : _push_messages)
            {
                _presist_messages.insert(make_pair(msgp->payload().attributes().id(), msgp));
            }
            _durable_total_size = _durable_valid_size = _presist_messages.size();
        }

        // 新增消息
        void insert_msg(const ns_proto::Attributes *attribute, const string &body,
                        DeliverMode durable)
        {
            // 1.构建MessagePtr对象
            MessagePtr msgp = make_shared<ns_proto::Message>();
            string id;
            ns_proto::DeliverMode deliver;
            string routing_key;
            if (attribute == nullptr)
            {
                id = Uuid::uuid();
                deliver = durable;
                routing_key = "";
            }
            else
            {
                id = attribute->id();
                deliver = attribute->del_mod();
                routing_key = attribute->routing_key();
            }
            msgp->mutable_payload()->mutable_attributes()->set_id(id);
            msgp->mutable_payload()->mutable_attributes()->set_del_mod(deliver);
            msgp->mutable_payload()->mutable_attributes()->set_routing_key(routing_key);
            msgp->mutable_payload()->set_body(body);

            // 2.判断是否进行持久化
            unique_lock<mutex> lock(_mtx); // 上面是创建对象，属于线程局部数据
            if (attribute->del_mod() == DeliverMode::DURABLE)
            {
                msgp->mutable_payload()->set_valid("1");
                _message_durable.insert_msg(msgp);                          // 添加进文件里
                _presist_messages.insert(make_pair(attribute->id(), msgp)); // 添加进内存的持久化表

                _durable_total_size++;
                _durable_valid_size++;
            }
            _push_messages.push_back(msgp); // 添加进内存的待推送链表
        }

        // 获取队首消息(用于发送)
        MessagePtr front()
        {
            // 获取队首消息时本质就是将待发送消息移到待确认消息中
            unique_lock<mutex> lock(_mtx);
            if (_push_messages.empty())
            {
                LOG("WARNING") << "该队列无消息，无法获取队首消息" << endl;
                return MessagePtr();
            }
            MessagePtr msgp = _push_messages.front();
            _push_messages.pop_front();
            _comfirm_messages.insert(make_pair(msgp->payload().attributes().id(), msgp));
            return msgp;
        }

        // 删除消息(用于处理待确认消息)
        void delete_msg(const string &msg_id)
        {
            // 1.查找待确认消息表中是否存在该条消息
            unique_lock<mutex> lock(_mtx);
            auto it = _comfirm_messages.find(msg_id);
            if (it == _comfirm_messages.end())
            {
                LOG(WARNING) << "该队列中不存在该消息" << endl;
                return;
            }
            _comfirm_messages.erase(msg_id);
            // 2.判断是否进行了持久化
            if (it->second->payload().attributes().del_mod() == DURABLE)
            {
                // 先删除文件中的消息，再删除内存中的消息
                _message_durable.delete_msg(_presist_messages[msg_id]);
                _presist_messages.erase(msg_id);
                _durable_valid_size--;

                // 此时"删除"了文件消息，则判断是否进行垃圾回收
                _gc();
            }
        }

        // 返回待推送消息数量
        size_t _push_size()
        {
            unique_lock<mutex> lock(_mtx);
            return _push_messages.size();
        }

        // 返回待确认消息数量
        size_t _comfirm_size()
        {
            unique_lock<mutex> lock(_mtx);
            return _comfirm_messages.size();
        }

        // 返回持久化消息数量
        size_t _presist_size()
        {
            unique_lock<mutex> lock(_mtx);
            return _presist_messages.size();
        }

        // 返回持久化总消息数量
        size_t _presist_total_size()
        {
            unique_lock<mutex> lock(_mtx);
            return _durable_total_size;
        }

        // 销毁队列数据与文件信息
        void clear()
        {
            unique_lock<mutex> lock(_mtx);
            _push_messages.clear();
            _comfirm_messages.clear();
            _presist_messages.clear();
            _message_durable.delete_msgfile();
            _durable_total_size = _durable_valid_size = 0;
        }

    private:
        void _gc()
        {
            // 当文件中有效消息不足总消息的一半，并且总消息数量超过1500 则进行回收
            if (_durable_total_size > 1500 && _durable_valid_size * 2 < _durable_total_size)
            {
                // 将文件的有效消息存放到链表中
                list<MessagePtr> tmp = _message_durable.recovery_and_gc();
                // 内存中对持久化消息的管理发生了offset的改变
                _presist_messages.clear(); // 直接清除重新push
                for (const auto &msgp : tmp)
                {
                    _presist_messages.insert(make_pair(msgp->payload().attributes().id(), msgp));
                }
                _durable_total_size = _durable_valid_size = _presist_messages.size();
            }
        }

    private:
        mutex _mtx;                                          // 互斥锁
        string _qname;                                       // 队列名称
        MessageDurable _message_durable;                     // 消息的文件管理
        list<MessagePtr> _push_messages;                     // 待推送消息链表
        unordered_map<string, MessagePtr> _comfirm_messages; // 待确认消息
        unordered_map<string, MessagePtr> _presist_messages; // 持久化消息

        // 确定文件消息是否进行垃圾回收
        size_t _durable_total_size; // 持久化(文件中)总消息数量
        size_t _durable_valid_size; // 持久化有效消息数量
    };

    // 对所有队列消息进行管理
    class MessageManage
    {
        using QueueMessagePtr = shared_ptr<QueueMessage>;

    public:
        MessageManage(const string &dir) : _path(dir)
        {
            // 创建队列文件存放路径
            assert(FileOper::createdir(_path));
        }

        // (初始化)创建、销毁消息队列
        void init_msg_queue(const string &qname)
        {
            QueueMessagePtr qmp;
            {
                // 锁是保护_msgs_map,而下面进行数据恢复的时候有其内部专门的锁进行保护
                // 构造的时候同时进行文件数据恢复，但是队列消息数据恢复操作，其内部有锁的保护
                // 所以将数据恢复操作独立起来，不让其在构造的时候进行恢复
                unique_lock<mutex> lock(_mtx);
                if (_msgs_map.count(qname))
                {
                    LOG(WARNING) << "该队列已存在" << endl;
                    return;
                }
                qmp = make_shared<QueueMessage>(_path, qname);
                _msgs_map.insert(make_pair(qname, qmp));
            }
            // 进行数据的恢复
            qmp->_recovery();
        }
        void destroy_msg_queue(const string &qname)
        {
            // 同时销毁内存数据和队列文件数据
            QueueMessagePtr qmp;
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "该队列消息不存在" << endl;
                    return;
                }
                qmp = _msgs_map[qname];
                _msgs_map.erase(qname);
            }
            // 销毁文件消息操作时，其内部存在锁保护
            qmp->clear();
        }

        // 向指定的消息队列插入信息
        void insert_msg(const string &qname, const ns_proto::Attributes *attribute,
                        const string &body, DeliverMode durable)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "向队列插入信息失败：不存在该队列" << endl;
                    return;
                }
            }
            _msgs_map[qname]->insert_msg(attribute, body, durable);
        }

        // 获取队首消息
        MessagePtr front(const string &qname)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "获取队首消息失败：不存在该队列" << endl;
                    return MessagePtr();
                }
            }
            return _msgs_map[qname]->front();
        }

        // 指定队列的消息确认(即删除确认消息表中的信息)
        void comfirm_msg(const string &qname, const string &msg_id)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "确认消息失败：不存在该队列" << endl;
                    return;
                }
            }
            _msgs_map[qname]->delete_msg(msg_id);
        }

        // 返回待推送消息数量
        size_t _push_size(const string &qname)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "返回推送消息数量失败：该队列消息不存在" << endl;
                    return 0;
                }
            }
            return _msgs_map[qname]->_push_size();
        }

        // 返回待确认消息数量
        size_t _comfirm_size(const string &qname)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "返回待确认消息数量失败：该队列不存在" << endl;
                    return 0;
                }
            }
            return _msgs_map[qname]->_comfirm_size();
        }

        // 返回持久化消息数量
        size_t _presist_size(const string &qname)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "返回持久化消息数量失败：该队列消息不存在" << endl;
                    return 0;
                }
            }
            return _msgs_map[qname]->_presist_size();
        }

        // 返回持久化总消息数量
        size_t _presist_total_size(const string &qname)
        {
            {
                unique_lock<mutex> lock(_mtx);
                if (!_msgs_map.count(qname))
                {
                    LOG(WARNING) << "返回持久化消息数量失败：该队列消息不存在" << endl;
                    return 0;
                }
            }
            return _msgs_map[qname]->_presist_total_size();
        }

        // --------------------------------------------测试
        // 返回管理的队列数量
        size_t qsize()
        {
            unique_lock<mutex> lock(_mtx);
            return _msgs_map.size();
        }
        // 返回管理的队列
        QueueMessagePtr qmp(const string &qname)
        {
            unique_lock<mutex> lock(_mtx);
            return _msgs_map[qname];
        }
        // 清理
        void clear()
        {
            unique_lock<mutex> lock(_mtx);
            for (auto &qmp : _msgs_map)
            {
                qmp.second->clear();
            }
        }

    private:
        mutex _mtx;
        string _path;                                     // 消息存放路径(用于每次创建新的消息队列时所配置的消息文件)
        unordered_map<string, QueueMessagePtr> _msgs_map; // 队列名称与队列消息的映射关系
    };
}